package com.varian.task.core;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.rabbitmq.client.Channel;
import com.varian.task.annotation.VarianTask;
import com.varian.task.constant.TaskConstant;
import com.varian.task.model.TaskUnprocessModel;
import com.varian.task.util.TaskUtil;
import com.varian.tool.Func;
import com.varian.tool.exception.ServerException;
import com.varian.tool.model.TaskStatus;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.Map;

/**
 * @author ben
 * @since 2024/11/21
 */
@Slf4j
public class TaskListenerContainerAdapter implements SmartInitializingSingleton {

    private final MessageConverter messageConverter;
    public final static Map<String, ITaskHandler> TASK_HANDLER_REPOSITORY = MapUtil.newConcurrentHashMap();

    public TaskListenerContainerAdapter(MessageConverter messageConverter) {
        this.messageConverter = messageConverter;
    }

    @Override
    public void afterSingletonsInstantiated() {
        ApplicationContext applicationContext = SpringUtil.getApplicationContext();
        // 初始化所有taskHandler
        initTaskHandlerMethod(applicationContext);
        // 初始化task监听器
        initTaskListener(applicationContext);
    }

    private void initTaskListener(ApplicationContext applicationContext) {
        SimpleRabbitListenerContainerFactory rabbitRegistrar = applicationContext.getBean(SimpleRabbitListenerContainerFactory.class);
        SimpleMessageListenerContainer listenerContainer = rabbitRegistrar.createListenerContainer();
        listenerContainer.setQueueNames(TaskConstant.QUEUE_TASK_EXECUTE);
        listenerContainer.setMessageListener((ChannelAwareMessageListener) this::onMessage);
        SpringUtil.registerBean("taskMessageListenerContainer", listenerContainer);
    }

    private void initTaskHandlerMethod(ApplicationContext applicationContext) {
        for (String beanName : applicationContext.getBeanDefinitionNames()) {
            Object bean = applicationContext.getBean(beanName);
            if (ObjUtil.isNotNull(bean)) {
                ReflectionUtils.doWithMethods(bean.getClass(), method -> {
                    VarianTask task = AnnotatedElementUtils.findMergedAnnotation(method, VarianTask.class);
                    if (ObjUtil.isNotNull(task)) {
                        registerTaskHandler(bean, method, task);
                    }
                });
            }
        }
    }

    public static ITaskHandler getTaskHandler(String taskCode) {
        return TASK_HANDLER_REPOSITORY.get(taskCode);
    }

    private void registerTaskHandler(Object bean, Method method, VarianTask task) {
        Assert.isFalse(TASK_HANDLER_REPOSITORY.containsKey(task.value()), () -> ServerException.of("task code [{}] 重复", task.value()));
        ITaskHandler taskHandler = resolveTaskHandler(bean, method, task);
        TASK_HANDLER_REPOSITORY.put(task.value(), taskHandler);
    }

    private ITaskHandler resolveTaskHandler(Object bean, Method method, VarianTask task) {
        Class<?> targetClass = bean.getClass();
        Method initMethod = null;
        if (StrUtil.isNotBlank(task.init())) {
            initMethod = ClassUtil.getDeclaredMethod(targetClass, task.init());
            Assert.notNull(initMethod, () -> ServerException.of("task code [{}] initMethod [{}] 不存在", task.value(), task.init()));
        }
        Method destroyMethod = null;
        if (StrUtil.isNotBlank(task.destroy())) {
            destroyMethod = ClassUtil.getDeclaredMethod(targetClass, task.destroy());
            Assert.notNull(destroyMethod, () -> ServerException.of("task code [{}] destroyMethod [{}] 不存在", task.value(), task.destroy()));
        }
        return new MethodTaskHandler(task.value(), bean, method, initMethod, destroyMethod);
    }

    @SneakyThrows
    private void onMessage(Message message, Channel channel) {
        Func.devOpt(() -> log.info("task message: {}", message));
        Object o = messageConverter.fromMessage(message);
        if (o instanceof TaskUnprocessModel task) {
            ITaskHandler taskHandler = getTaskHandler(task.getTaskCode());
            if (ObjUtil.isNotNull(taskHandler)) {
                doTaskExecute(taskHandler, task);
            } else {
                log.warn("task code [{}] not found", task.getTaskCode());
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            }
        }
    }

    private void doTaskExecute(ITaskHandler taskHandler, TaskUnprocessModel task) {
        LocalDateTime now = LocalDateTime.now();
        TaskStatus taskStatus = TaskStatus.SUCCESS;
        String errorMessage = null;
        try {
            taskHandler.execute(task);
        } catch (Exception e) {
            taskStatus = TaskStatus.FAILURE;
            errorMessage = e.getCause().getMessage();
            log.error("task code [{}] execute error", task.getTaskCode(), e);
        } finally {
            TaskUtil.updateTaskStatus(task.getTaskId(), taskStatus, now, errorMessage);
        }
    }
}
